package com.eight.cloud.rabbitmq.config;

import com.eight.cloud.core.global.exception.MessageSendException;
import com.eight.cloud.core.global.exception.MessageTooLargeException;
import com.eight.cloud.rabbitmq.util.SendStrategy;
import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.units.qual.C;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.lang.NonNull;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.MimeTypeUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 消息中间件工具类
 *
 * @ClassName: MessageQueueProducer
 * @Author: TXC
 * @Date: 2025-08-13 14:26
 **/
@Slf4j
@Component
public class RabbitMqPublisher {

    @Value("${spring.application.name}")
    private String applicationName;
    @Resource
    private StreamBridge streamBridge;
    @Resource
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    /**
     * 消息ID
     */
    public static final String MESSAGE_ID = "messageId";
    /**
     * 发送应用
     */
    public static final String SOURCE_APP = "sourceApp";
    /**
     * 跟踪ID
     */
    public static final String TRACE_ID = "traceId";

    /**
     * 路由键
     */
    public static final String ROUTING_KEY = "routingKey";
    /**
     * content type
     */
    public static final String CONTENT_TYPE = MessageHeaders.CONTENT_TYPE;
    /**
     * 最大消息大小（字节）
     */
    private static final int MAX_MESSAGE_SIZE_BYTES = 1_048_576; // 1MB
    /**
     * 活动状态
     */
    private final AtomicBoolean isActive = new AtomicBoolean(true);


    @PostConstruct
    public void initialize() {
        log.info("RabbitMQ Producer initialized for application: {}", applicationName);
    }

    @PreDestroy
    public void shutdownGracefully() {
        isActive.set(false);
        log.info("Initiating graceful shutdown for RabbitMQ Producer...");
        // 等待任务完成
        threadPoolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        threadPoolTaskExecutor.setAwaitTerminationSeconds(30);
        threadPoolTaskExecutor.shutdown();
        log.info("RabbitMQ Producer shutdown completed");
    }

    /**
     * 发送消息（基础方法）
     *
     * @param bindingName 绑定目标名称
     * @param payload     消息内容
     * @return 发送是否成功
     */
    public boolean send(@NonNull String bindingName, @NonNull Object payload) {
        return send(bindingName, buildNewMessage(payload));
    }

    /**
     * 发送消息（基础方法）
     *
     * @param bindingName 绑定目标名称
     * @param payload     消息内容
     * @return 发送是否成功
     */
    public boolean send(@NonNull String bindingName, @Nullable String routingKey, @NonNull Object payload) {
        Map<String, Object> headers = new HashMap<>();
        headers.put(ROUTING_KEY, routingKey);
        return send(bindingName, buildNewMessage(payload, headers));
    }

    /**
     * 发送自定义消息
     *
     * @param bindingName 绑定目标名称
     * @param message     完整消息对象
     * @return 发送是否成功
     */
    public boolean send(@NonNull String bindingName, @NonNull Message<?> message) {
        return sendWithStrategy(bindingName, message, SendStrategy.defaultStrategy());
    }

    /**
     * 发送消息（带发送策略）
     *
     * @param bindingName 绑定目标名称
     * @param payload     消息内容
     * @param strategy    发送策略配置
     * @return 发送是否成功
     */
    public boolean send(@NonNull String bindingName, @NonNull String routingKey, @NonNull Object payload, @NonNull SendStrategy strategy) {
        Map<String, Object> headers = new HashMap<>();
        headers.put(ROUTING_KEY, routingKey);
        return sendWithStrategy(bindingName, buildNewMessage(payload, headers), strategy);
    }

    /**
     * 发送消息（带发送策略）
     *
     * @param bindingName 绑定目标名称
     * @param payload     消息内容
     * @param strategy    发送策略配置
     * @return 发送是否成功
     */
    public boolean send(@NonNull String bindingName, @NonNull Object payload, @NonNull SendStrategy strategy) {
        return sendWithStrategy(bindingName, buildNewMessage(payload), strategy);
    }

    /**
     * 异步发送消息
     *
     * @param bindingName 绑定目标名称
     * @param payload     消息内容
     * @return CompletableFuture 发送结果
     */
    public CompletableFuture<Boolean> sendAsync(@NonNull String bindingName, @NonNull Object payload) {
        return sendAsync(bindingName, buildNewMessage(payload));
    }

    /**
     * 异步发送自定义消息
     *
     * @param bindingName 绑定目标名称
     * @param message     完整消息对象
     * @return CompletableFuture 发送结果
     */
    public CompletableFuture<Boolean> sendAsync(@NonNull String bindingName, @NonNull Message<?> message) {
        return sendAsync(bindingName, message, SendStrategy.defaultStrategy());
    }

    /**
     * 异步发送消息（带发送策略）
     *
     * @param bindingName 绑定目标名称
     * @param message     消息对象
     * @param strategy    发送策略配置
     * @return CompletableFuture 发送结果
     */
    public CompletableFuture<Boolean> sendAsync(@NonNull String bindingName,
                                                @NonNull Message<?> message,
                                                @NonNull SendStrategy strategy) {
        ensureProducerActive();
        return CompletableFuture.supplyAsync(() ->
                        sendWithStrategy(bindingName, message, strategy),
                threadPoolTaskExecutor
        );
    }

    /**
     * 创建新消息构建器
     *
     * @param payload 消息内容
     * @return 消息构建器实例
     */
    public MessageBuilder<?> messageBuilder(@NonNull Object payload) {
        return MessageBuilder.withPayload(payload)
                .setHeader(MESSAGE_ID, UUID.randomUUID().toString())
                .setHeader(SOURCE_APP, applicationName)
                .setHeader(TRACE_ID, generateTraceId())
                .setHeader(CONTENT_TYPE, detectContentType(payload));
    }

    /**
     * 构建新消息
     *
     * @param payload 消息内容
     * @return 消息对象
     */
    public Message<?> buildNewMessage(@NonNull Object payload) {
        return messageBuilder(payload).build();
    }

    /**
     * 构建带自定义头的消息
     *
     * @param payload 消息内容
     * @param headers 自定义消息头
     * @return 消息对象
     */
    public Message<?> buildNewMessage(@NonNull Object payload, @Nullable Map<String, Object> headers) {
        MessageBuilder<?> builder = messageBuilder(payload);
        if (headers != null) {
            headers.forEach(builder::setHeader);
        }
        return builder.build();
    }


    /**
     * 发送消息（带发送策略）
     *
     * @param bindingName 绑定目标名称
     * @param message     消息对象
     * @param strategy    发送策略
     * @return 发送是否成功
     */
    private boolean sendWithStrategy(@NonNull String bindingName,
                                     @NonNull Message<?> message,
                                     @NonNull SendStrategy strategy) {
        ensureProducerActive();
        validateMessageSize(message);
        try {
            // 事务感知处理
            if (strategy.isTransactionAware() && TransactionSynchronizationManager.isSynchronizationActive()) {
                return handleTransactionalSend(bindingName, message);
            }
            // 应用速率限制
            if (strategy.getRateLimiter() != null && !strategy.getRateLimiter().tryAcquire()) {
                log.warn("Rate limit exceeded for binding: {}", bindingName);
                return false;
            }
            // 执行带重试的发送
            return executeWithRetry(bindingName, message, strategy);
        } catch (Exception e) {
            throw new MessageSendException("Failed to send message to " + bindingName, e);
        }
    }

    /**
     * 处理事务感知发送
     *
     * @param bindingName 绑定目标名称
     * @param message     消息对象
     * @return 发送是否成功
     */
    private boolean handleTransactionalSend(String bindingName, Message<?> message) {
        log.debug("Registering message for transactional send: {}", message.getHeaders().get(MESSAGE_ID));
        TransactionSynchronizationManager.registerSynchronization(
                new TransactionSynchronization() {
                    @Override
                    public void afterCommit() {
                        log.debug("Executing transactional send after commit");
                        executeWithRetry(bindingName, message, SendStrategy.defaultStrategy());
                    }
                }
        );
        return true;
    }

    /**
     * 执行带重试的发送
     *
     * @param bindingName 绑定目标名称
     * @param message     消息对象
     * @param strategy    发送策略
     * @return 发送是否成功
     */
    private boolean executeWithRetry(String bindingName, Message<?> message, SendStrategy strategy) {
        int attempt = 0;
        long delay = strategy.getInitialRetryDelay();
        while (attempt < strategy.getMaxRetries() && isActive.get()) {
            attempt++;
            if (attempt > 1) {
                log.info("Retry attempt {}/{} for binding: {}", attempt, strategy.getMaxRetries(), bindingName);
            }
            try {
                boolean success = streamBridge.send(bindingName, message);
                if (success) {
                    log.debug("Message sent successfully to {}", bindingName);
                    return true;
                }
                log.warn("Send operation returned false for binding: {}", bindingName);
            } catch (Exception ex) {
                log.error("Send failed for binding: {}, attempt: {}/{}", bindingName, attempt,
                        strategy.getMaxRetries(), ex);
            }
            // 等待后重试
            if (attempt < strategy.getMaxRetries() && delay > 0) {
                sleepForRetry(delay);
                // 指数退避
                if (strategy.isExponentialBackoff()) {
                    delay *= 2;
                }
            }
        }
        log.error("All send attempts failed for binding: {}", bindingName);
        return false;
    }

    /**
     * 确保生产者活动
     *
     * @throws IllegalStateException 如果生产者已关闭
     */
    private void ensureProducerActive() {
        if (!isActive.get()) {
            throw new IllegalStateException("Producer is shutting down and not accepting new messages");
        }
    }

    /**
     * 验证消息大小是否超过最大限制
     *
     * @param message 消息对象
     */
    private void validateMessageSize(Message<?> message) {
        int size = calculateMessageSize(message);
        if (size > MAX_MESSAGE_SIZE_BYTES) {
            throw new MessageTooLargeException(
                    "Message size " + size + " bytes exceeds maximum of " + MAX_MESSAGE_SIZE_BYTES + " bytes");
        }
    }

    /**
     * 计算消息大小（字节数）
     *
     * @param message 消息对象
     * @return 消息大小（字节数）
     */
    private int calculateMessageSize(Message<?> message) {
        Object payload = message.getPayload();
        if (payload instanceof byte[]) {
            return ((byte[]) payload).length;
        }
        if (payload instanceof String) {
            return ((String) payload).getBytes().length;
        }
        // 其他类型不计算大小
        return 0;
    }

    /**
     * 检测消息内容类型
     *
     * @param payload 消息内容
     * @return 消息内容类型
     */
    private String detectContentType(Object payload) {
        if (payload instanceof String || payload instanceof byte[]) {
            return MimeTypeUtils.TEXT_PLAIN_VALUE;
        }
        return MimeTypeUtils.APPLICATION_JSON_VALUE;
    }

    /**
     * @return 生成的TraceId
     */
    private String generateTraceId() {
        return UUID.randomUUID().toString();
    }

    /**
     * 等待指定时间后重试
     *
     * @param milliseconds 重试等待时间（毫秒）
     */
    private void sleepForRetry(long milliseconds) {
        try {
            Thread.sleep(milliseconds);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}
